<?php
/**
 * mq mq
 * @package common_lib_mq
 */
load_lib('/db/pooling');
load_lib('/sys/variable');
/**
 * mq mq
 * @author pandajingjing
 * @package common_lib_mq
 */
class mq_mq{

	/**
	 * mq通道实例
	 * @var object
	 */
	private $_objChannel='';

	/**
	 * 当前队列
	 * @var string
	 */
	private $_strCurrentQueue='';

	/**
	 * 发送绑定池
	 * @var array
	 */
	private $_arrSendStmt=array();

	/**
	 * 获取绑定池
	 * @var array
	 */
	private $_arrGetStmt=array();

	/**
	 * 删除绑定池
	 * @var array
	 */
	private $_arrFadeStmt=array();

	/**
	 * 批量获取队列数量
	 * @var int
	 */
	private $_intMsgConsumeCnt=100;

	/**
	 * 构造函数
	 * @param string $p_strDSN
	 */
	function __construct($p_strChannel){
		$this->_objChannel=db_pooling::getInstance()->getConnect($p_strChannel);
	}

	/**
	 * 绑定队列
	 * @param string $p_strQueue
	 */
	function bindQueue($p_strQueue){
		$this->_strCurrentQueue=$p_strQueue;
	}

	/**
	 * 发送消息
	 * @param mix $p_mixData
	 */
	function sendMessage($p_mixData){
		if(isset($this->_arrSendStmt[$this->_strCurrentQueue])){

		}else{
			$strSQL='insert into '.$this->_strCurrentQueue.' (QDATA,DSTATUS,CREATETIME)values(:QDATA,:DSTATUS,:CREATETIME)';
			$this->_arrSendStmt[$this->_strCurrentQueue]=$this->_objChannel->prepare($strSQL);
		}
		$this->_arrSendStmt[$this->_strCurrentQueue]->execute(array( 
				':QDATA'=>json_encode($p_mixData),
				':DSTATUS'=>1,
				':CREATETIME'=>sys_variable::getInstance()->getTime() 
		));
	}

	/**
	 * 消费消息
	 * @param mix $p_mixUserFunc
	 */
	function consumeMessage($p_mixUserFunc){
		if(isset($this->_arrGetStmt[$this->_strCurrentQueue])){

		}else{
			$strSQL='select QID,QDATA from '.$this->_strCurrentQueue.' where DSTATUS=:DSTATUS order by QID asc limit '.$this->_intMsgConsumeCnt;
			$this->_arrGetStmt[$this->_strCurrentQueue]=$this->_objChannel->prepare($strSQL);
			$this->_arrGetStmt[$this->_strCurrentQueue]->setFetchMode(PDO::FETCH_ASSOC);
		}
		$this->_arrGetStmt[$this->_strCurrentQueue]->execute(array( 
				':DSTATUS'=>1 
		));
		$mixData=$this->_arrGetStmt[$this->_strCurrentQueue]->fetchAll();
		if(0==count($mixData)){
			//sleep(1);
		}else{
			if(isset($this->_arrFadeStmt[$this->_strCurrentQueue])){

			}else{
				$strSQL='update '.$this->_strCurrentQueue.' set DSTATUS=:DSTATUS, UPDATETIME=:UPDATETIME where QID=:QID';
				$this->_arrFadeStmt[$this->_strCurrentQueue]=$this->_objChannel->prepare($strSQL);
			}
			foreach($mixData as $arrData){
				call_user_func($p_mixUserFunc,new mq_msg($arrData['QDATA'],$arrData['QID'],$this->_arrFadeStmt[$this->_strCurrentQueue]));
			}
		}
	}
}

/**
 * mq msg
 * @author pandajingjing
 * @package common_lib_mq
 */
class mq_msg{

	/**
	 * 消息体
	 * @var mix
	 */
	private $_mixBody='';

	/**
	 * 消息队列号
	 * @var int
	 */
	private $_intMsgID=0;

	/**
	 * 消息删除绑定
	 * @var object
	 */
	private $_objFadeStmt='';

	/**
	 * 构造函数
	 * @param string $p_strBody
	 * @param int $p_intMsgID
	 * @param object $p_objFadeStmt
	 */
	function __construct($p_strBody,$p_intMsgID,$p_objFadeStmt){
		$this->_mixBody=json_decode($p_strBody,true);
		$this->_intMsgID=$p_intMsgID;
		$this->_objFadeStmt=$p_objFadeStmt;
	}

	/**
	 * 获取消息体内容
	 * @return mix
	 */
	function getBody(){
		return $this->_mixBody;
	}

	/**
	 * 删除消息队列
	 */
	function fade(){
		$this->_objFadeStmt->execute(array( 
				':QID'=>$this->_intMsgID,
				':DSTATUS'=>2,
				':UPDATETIME'=>sys_variable::getInstance()->getTime() 
		));
	}

}